package edu.ucla.cs.rpc.multicast.destination.agreement;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.List;
import java.util.Set;

import edu.ucla.cs.rpc.multicast.Dest;
import edu.ucla.cs.rpc.multicast.handlers.MessageHandler;
import edu.ucla.cs.rpc.multicast.network.MulticastMessageReceiver;
import edu.ucla.cs.rpc.multicast.network.MulticastMessageSender;
import edu.ucla.cs.rpc.multicast.network.message.Message;
import edu.ucla.cs.rpc.multicast.util.IdentifierComparator;

/**
 * 
 * @author Philip Russell
 * @see Total Order Broadcast and Multicast Algorithms: Taxonomy and Survery,
 *      Defago et al
 * @see Section 4.5, pages 18 - 19
 * 
 */
public class AgreementDest implements Dest {

	private class MessageData {
		protected Set<Long> senderIDSet;

		protected long maxTimestamp;

		public MessageData(long id, long timestamp) {
			this.senderIDSet = new HashSet<Long>();
			this.senderIDSet.add(id);
			this.maxTimestamp = timestamp;
		}
	}

	private Hashtable<Long, MessageData> messageTable = null;

	private MessageHandler handler;

	private long logicalClock;

	private Set<Message> received = null;

	private Set<Message> stamped = null;

	private MulticastMessageReceiver receiver = null;

	private MulticastMessageSender sender = null;

	public AgreementDest(SocketAddress manager, MessageHandler handler)
			throws IOException {

		this.handler = handler;

		this.receiver = new MulticastMessageReceiver(manager,
				new AgreementDestMessageHandler());

		this.sender = new MulticastMessageSender(manager);

		this.messageTable = new Hashtable<Long, MessageData>();

		/* CODE FROM SECTION 4, FIGURE 14 */

		this.stamped = new HashSet<Message>();
		this.received = new HashSet<Message>();
		this.logicalClock = 0;
	}

	private class AgreementDestMessageHandler implements MessageHandler {

		public void receive(Message message) {

			try {
				if (message.getTimestamp() == Message.NO_TIMESTAMP) {
					message.setTimestamp(logicalClock);

					// this.sender.send(rpcMessage.toPacket());
					sender.send(message);

					assert (!messageTable.containsKey(message.getSeqNum()));
					messageTable.put(message.getSeqNum(), new MessageData(
							message.getSenderID(), message.getTimestamp()));

					received.add(message);
					logicalClock++;

				} else {

					logicalClock = Math.max(logicalClock, message
							.getTimestamp());

					/* Update the information about the message in the table */
					MessageData info = messageTable.get(message.getSeqNum());
					assert (info != null);
					info.senderIDSet.add(message.getSenderID());
					info.maxTimestamp = Math.max(info.maxTimestamp, message
							.getTimestamp());

					/*
					 * TODO: Figure out how to extract ONLY the set of RECEIVERs
					 * from a MulticastMessageReceiver, instead of the SENDERs
					 * as well One possiblity is to have two MesageReceivers on
					 * differenet ports.
					 * UPDATE: ain't gonna happen
					 */
					
					Set<Long> receiverSet = receiver.getReceivers();
					Set<Long> messageSenderSet = info.senderIDSet;

					if (messageSenderSet.containsAll(receiverSet)) {

						long globalSeqnum = info.maxTimestamp;

						Message globalStampedMessage = new Message(message);
						globalStampedMessage.setSeqNum(globalSeqnum);
						stamped.add(globalStampedMessage);

						received.remove(message);
						List<Message> deliverable = new ArrayList<Message>();

						for (Message globalMessage : stamped) {
							for (Message recMessage : received) {
								if (globalMessage.getTimestamp() < recMessage
										.getTimestamp()) {
									deliverable.add(globalMessage);
								}
							}
						}
						/* SORT DELIVERABLE */
						Collections.sort(deliverable,
								new IdentifierComparator());

						/*
						 * DELIVER MESSAGES IN DELIVERABLE TO DESTINATION
						 * HANDLER
						 */
						for (Message deliverableMessage : deliverable) {
							handler.receive(deliverableMessage);
						}

						/* TODO : update the sets? */

						/* STAMPED = SET DIFFERENCE OF STAMPED , DELIVERABLE */
						stamped.removeAll(deliverable);
					}
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	public void shutdown() {
		receiver.shutdown();
		sender.shutdown();
	}

	public void setMessageHandler(MessageHandler handler) {
		this.handler = handler;
	}

}
